package com.cmb.lk50.config;

import com.cmb.lk50.common.AccessInterceptor;
import com.cmb.lk50.common.ServletWrapperFilter;
import com.cmb.lk50.common.WebFluxAccessFilter;
import com.cmb.lk50.sink.LogSink;
import com.cmb.lk50.sink.console.ConsoleSink;
import com.cmb.lk50.sink.db.JdbcSink;
import com.cmb.lk50.sink.db.StatementGenerator;
import com.cmb.lk50.sink.es.EsSink;
import com.cmb.lk50.sink.kafka.KafkaSink;
import com.cmb.lk50.sink.mongo.MongoSink;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.gson.Gson;
import com.google.gson.JsonElement;
import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoClientURI;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;
import io.searchbox.client.JestClientFactory;
import io.searchbox.client.config.HttpClientConfig;
import org.apache.catalina.startup.Tomcat;
import org.apache.coyote.UpgradeProtocol;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.common.security.plain.PlainLoginModule;
import org.apache.kafka.common.serialization.StringSerializer;
import org.bson.codecs.Codec;
import org.bson.codecs.configuration.CodecRegistries;
import org.bson.codecs.configuration.CodecRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;
import org.springframework.web.server.WebFilter;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;

import javax.annotation.Resource;
import javax.servlet.Filter;
import javax.servlet.Servlet;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;

/**
 * 自动配置类
 * Author       : 蒋俊钊 80256973
 * Date         : 2020/7/16
 * Copyright    (C) ChinaMerchantsBank
 */
@Configuration
@ConditionalOnProperty(prefix = "access-log", name = "enable", havingValue = "true")
@ConditionalOnWebApplication
@EnableConfigurationProperties({LogInProperties.class})
public class LogInAutoConfiguration implements ApplicationContextAware {

    private static final Logger LOGGER = LoggerFactory.getLogger(LogInAutoConfiguration.class);

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 配置Servlet环境下的拦截器和过滤器
     * //@ServletComponentScan("com.cmb.lk50.common")
     * 注意: 要配置成静态类 或 添加配置条件
     */
    @Configuration
    @ConditionalOnClass({Servlet.class, Tomcat.class, UpgradeProtocol.class})
    @ConditionalOnProperty(prefix = "access-log", name = "enable", havingValue = "true")
    public class ServletInterceptorConfig implements WebMvcConfigurer {

        @Bean
        public Filter wrapperFilter() {
            LOGGER.info("[LogIn] - 配置Web过滤器");
            return new ServletWrapperFilter();
        }

        @Bean
        public AccessInterceptor accessInterceptor() {
            LOGGER.info("[LogIn] - 配置MVC拦截器");
            return new AccessInterceptor();
        }

        // 拦截所有请求，在拦截器中根据注解AccessLog过滤
        @Override
        public void addInterceptors(InterceptorRegistry registry) {
            registry.addInterceptor(accessInterceptor()).addPathPatterns("/**");
        }
    }

    /**
     * 配置Webflux环境下的拦截器
     * conditional条件很重要，不能用OnMissingBean(ServletInterceptorConfig.class), 否则会触发webmvc中的类加载
     */
    @Configuration
    @ConditionalOnMissingClass({"javax.servlet.Servlet", "org.apache.catalina.startup.Tomcat"})
    @ConditionalOnProperty(prefix = "access-log", name = "enable", havingValue = "true")
    public class WebFluxInterceptorConfig {

        @Bean
        public WebFilter accessFilter() {
            LOGGER.info("[LogIn] - 配置Webflux拦截器");
            return new WebFluxAccessFilter();
        }
    }


    /**
     * 通用配置，配置各种sink
     */
    @Configuration
    public class SinksConfig {

        @Resource
        private LogInProperties logInProperties;

        @Autowired(required = false)
        private ExecutorService asyncExecutor;


        @Autowired(required = false)
        private List<StatementGenerator.LogMapper<?>> logMappers;

        // todo 重构 BeanDefinition
        @Bean
        public List<LogSink> sinks() {
            ArrayList<LogSink> logSinks = new ArrayList<>();
            for (Map.Entry<String, Object> entry : logInProperties.getSinks().entrySet()) {
                String key = entry.getKey();
                if ("console".equals(key)) {
                    if ((Boolean) entry.getValue()) {
                        logSinks.add(new ConsoleSink());
                        LOGGER.info("[LogIn] - 配置Console Sink");
                    }
                } else if ("elasticsearch".equals(key)) {
                    Gson gson = new Gson();
                    JsonElement jsonElement = gson.toJsonTree(entry.getValue());
                    LogInProperties.EsProperties esProperties = gson.fromJson(jsonElement, LogInProperties.EsProperties.class);
                    JestClientFactory factory = new JestClientFactory();

                    // Jest http client
                    Set<String> uris = Arrays.stream(esProperties.getUris().split(","))
                      .map(s -> {
                          if (!s.startsWith("http://")) { s = "http://" + s; }
                          return s;
                      })
                      .collect(Collectors.toSet());
                    HttpClientConfig.Builder builder = new HttpClientConfig.Builder(uris);
                    if (StringUtils.hasText(esProperties.getUsername())) {
                        builder.defaultCredentials(esProperties.getUsername(), esProperties.getPassword());
                    }
                    builder.multiThreaded(esProperties.isMultiThreaded())
                      .connTimeout(esProperties.getConnectionTimeout())
                      .readTimeout(esProperties.getReadTimeout());

                    factory.setHttpClientConfig(builder.build());
                    logSinks.add(new EsSink(factory.getObject(), esProperties.getDefaultIndex(), esProperties.getDefaultType(),
                      esProperties.getUsername(), esProperties.getPassword(), asyncExecutor));
                    LOGGER.info("[LogIn] - 配置ElasticSearch Sink");
                } else if ("jdbc".equals(key)) {
                    Gson gson = new Gson();
                    JsonElement jsonElement = gson.toJsonTree(entry.getValue());
                    LogInProperties.JdbcProperties jdbcProperties = gson.fromJson(jsonElement, LogInProperties.JdbcProperties.class);
                    HikariConfig hikariConfig = new HikariConfig();
                    hikariConfig.setJdbcUrl(jdbcProperties.getUrl());
                    hikariConfig.setDriverClassName(jdbcProperties.getDriverClassName());
                    hikariConfig.setUsername(jdbcProperties.getUsername());
                    hikariConfig.setPassword(jdbcProperties.getPassword());
                    hikariConfig.setMinimumIdle(2);
                    hikariConfig.setMaximumPoolSize(5);
                    hikariConfig.setPoolName("LogIn-JDBC-Pool");

                    StatementGenerator statementGenerator = new StatementGenerator();
                    if (logMappers != null) {
                        logMappers.forEach(statementGenerator::addMapper);
                    }

                    logSinks.add(new JdbcSink(new HikariDataSource(hikariConfig), asyncExecutor, statementGenerator));
                    LOGGER.info("[LogIn] - 配置JDBC Sink");
                } else if ("kafka".equals(key)) {
                    Map<String, Object> props = (Map<String, Object>) entry.getValue();
                    if (props.get("username") != null && props.get("password") != null) {
                        props.put("sasl.jaas.config", PlainLoginModule.class.getName()
                          + " required username=\"" + props.remove("username") + "\" pass" + "word=\"" + props.remove("password") + "\";");
                        props.put("security.protocol", "SASL_PLAINTEXT");
                        props.put("sasl.mechanism", "PLAIN");
                    }
                    String topic = (String) props.remove("defaultTopic");
                    logSinks.add(new KafkaSink(new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()), topic));
                    LOGGER.info("[LogIn] - 配置Kafka Sink");
                } else if ("mongodb".equals(key)) {
                    Gson gson = new Gson();
                    JsonElement jsonElement = gson.toJsonTree(entry.getValue());
                    LogInProperties.MongodbProperties mongodbProperties = gson.fromJson(jsonElement, LogInProperties.MongodbProperties.class);

                    Map<String, Codec> codecsMap = applicationContext.getBeansOfType(Codec.class);
                    ArrayList<Codec<?>> codecs = new ArrayList<>();
                    for (Codec value : codecsMap.values()) {
                        codecs.add(value);
                    }
                    CodecRegistry codecRegistry = CodecRegistries.fromCodecs(codecs);
                    MongoClientOptions.Builder builder = new MongoClientOptions.Builder()
                      .codecRegistry(codecRegistry);

                    MongoClient mongoClient = new MongoClient(new MongoClientURI(mongodbProperties.getUri(), builder));
                    logSinks.add(new MongoSink(mongoClient, mongodbProperties, asyncExecutor));
                    LOGGER.info("[LogIn] - 配置Mongo Sink");
                } else {
                    LOGGER.warn("[LogIn] - 未知配置[{}]", key);
                }
            }
            return logSinks;
        }
    }

    /**
     * 通用配置，配置独立的异步线程池
     */
    @Configuration
    public class AsyncConfig {

        @Bean
        public ExecutorService asyncExecutor() {
            LOGGER.info("[LogIn] - 配置异步线程池");
            ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("LogIn-Pool-%d").build();
            return Executors.newCachedThreadPool(threadFactory);
        }
    }

}
